package cn.doitedu.zk.sensor;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * @date: 2019/7/8
 * @site: www.doitedu.cn
 * @author: hunter.d 涛哥
 * @qq: 657270652
 * @description: 动态感知系统的客户端实现
 */
public class Consumer {

    ZooKeeper zk = null;
    ArrayList<String> servers = new ArrayList<>();


    /**
     * 构造zk连接，并传入事件通知的回调函数
     *
     * @throws Exception
     */
    public void connectZookeeper() throws Exception {

        zk = new ZooKeeper("spark01:2181,spark02:2181,spark03:2181", 2000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {

                if (event.getType() == Event.EventType.NodeChildrenChanged) {
                    // 再次查询在线服务器列表
                    // 再次注册监听
                    try {
                        getOnlineServers();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });


    }


    /**
     * 查询zk中的/servers下的子节点，获得在线服务器信息，并注册监听
     */
    public void getOnlineServers() throws Exception {

        List<String> children = zk.getChildren("/servers", true);

        ArrayList<String> lst = new ArrayList<>();

        for (String child : children) {
            byte[] data = zk.getData("/servers/" + child, false, null);
            String serverInfo = new String(data);
            lst.add(serverInfo);
        }

        servers = lst;
        System.out.println("查询了一次服务器列表，结果如下： " + servers);

    }


    /**
     * 处理业务
     */
    public void requestService() throws IOException, InterruptedException {

        while (true) {
            Random random = new Random();
            int i = random.nextInt(servers.size());
            String serverInfo = servers.get(i);

            String[] split = serverInfo.split(":");

            System.out.println("挑选了 " + split[0] + "准备进行业务请求");


            Socket sc = new Socket(split[0], Integer.parseInt(split[1]));

            OutputStream out = sc.getOutputStream();
            InputStream in = sc.getInputStream();

            out.write("你好吗?".getBytes());
            out.flush();

            byte[] buf = new byte[1024];
            int read = in.read(buf);
            System.out.println("收到服务器端的响应结果： " + new String(buf, 0, read));

            in.close();
            out.close();
            sc.close();

            Thread.sleep(2000);

        }
    }


    public static void main(String[] args) throws Exception {

        Consumer consumer = new Consumer();

        // 连接zookeeper
        consumer.connectZookeeper();


        // 查询在线的服务器,并注册监听
        consumer.getOnlineServers();

        // 挑选在线的服务器，干业务去了
        consumer.requestService();

        consumer.zk.close();
    }


}
